package tool;

import bean.Score;
import com.google.common.collect.Sets;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import joinery.DataFrame;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.junit.experimental.theories.DataPoint;

import java.util.*;

import static tool.TableProperty.*;

public class StaticProvider {

    private static StaticProvider instance;
    private StaticProvider(){}
    public static StaticProvider getInstance()
    {
        if(instance == null)
        {
            instance = new StaticProvider();
        }
        return instance;
    }

    public static final Set<String> NUM_COL = Sets.newHashSet(SCORE, CREDIT, AGE, CLASS);
    public static DataFrame<Object> provideSC()
    {
        final List<String> cHeader = Arrays.asList(ID, CID);
        DataFrame<Object> classes = new DataFrame<>();
        classes.add(ID,SID,CID,SCORE,TID,TYPE);

        Observable<ResultScanner> cResult = DataProvider.catchAllFromType(SC);
        mapRData(SC,cHeader,classes,cResult,false);
        return classes;
    }

    /**
     * 所有学生的选课数目排名
     * @return ID CID(选课计次)
     */
    public DataFrame provideStudentChooseClassCountRank()
    {
        DataFrame<String> result = DataProvider.loadAllFromType(SC)
                .drop(SCORE,TID,TYPE,SID)
                .groupBy(ID)
                .count()
                .sortBy(CID);
        return result;
    }

    /**
     *  所有学生的平均成绩排名
     * @return ID SCORE
     */
    public DataFrame provideStudentMeanScoreRank()
    {
        DataFrame<Object> result = provideSC().groupBy(ID)
                .mean()
                .sortBy(SCORE);
        return result;
    }


    /**
     *  所有课程的选课人数的柱状图
     * @return CID,ID
     */
    public DataFrame provideClassChoosedCountRank()
    {
        DataFrame<String> result = DataProvider.loadAllFromType(SC)
                .drop(SID,SCORE,TID,TYPE)
                .groupBy(CID)
                .count()
                .sortBy(ID);
        return result;
    }

    /**
     * 所有课程的平均成绩排名课程
     * @return CID SCORE
     */
    public DataFrame provideClassMeanScoreRank()
    {
        DataFrame<Object> result = StaticProvider.provideSC()
//                .fillna()
                .groupBy(CID)
                .mean()
                .sortBy(SCORE);

        return result;
    }

    public DataFrame provideAClassStudentScoreRank(final String cid)
    {
        DataFrame<Object> lists = StaticProvider.provideSC()
                .select(new DataFrame.Predicate<Object>() {
                    @Override
                    public Boolean apply(List<Object> objects) {
//                        Lg.info(objects);
                        return objects.get(2).toString().contains(cid);
                    }
                })
                .groupBy(CID)
                .drop(TID,TYPE,ID)
                .sortBy(CID, SCORE);
        return lists;
    }

    /**
     * 所有课程最大成绩排名
     * @return CID SCORE
     */
    public DataFrame provideClassMaxScoreRank()
    {
        DataFrame<Object> result = StaticProvider.provideSC()
                .groupBy(CID,SID)
                .max()
                .sortBy(CID,SCORE);
        return result;
    }

    /**
     * 列出所有学生的年龄分布
     * @return
     */
    public DataFrame provideStudentAgeCount()
    {
        DataFrame<Object> result = StaticProvider.loadAllFromType(S)
                .drop(SID,NAME,SEX,BIRTHDAY,DNAME,CLASS,TYPE)
                .groupBy(AGE)
                .count()
                .sortBy(ID);
        return result;
    }

    /**
     * 每位同学的成绩分布：优秀、良好、合格、不合格的课程门数
     * @param sid
     * @return
     */
    public DataFrame provideAStudentScoreComment(final String sid)
    {
        DataFrame<Object> count = provideSC()
                .select(new DataFrame.Predicate<Object>() {
                    @Override
                    public Boolean apply(List<Object> objects) {
                        return objects.get(1).toString().contains(sid);
                    }
                })
                .groupBy(new DataFrame.KeyFunction<Object>() {
                    @Override
                    public Object apply(List<Object> objects) {
                        List list = new LinkedList();
                        list.add(objects.get(1));
                        list.add(new Score(Integer.parseInt(objects.get(3).toString())).getStar());
                        return list;
                    }
                })
                .count()
                .drop(ID,SID,CID,TID,TYPE);
        return count;
    }

    public static void main(String[] args) {

        DataFrame<Object> lists = getInstance().provideClassMaxScoreRank();
        System.out.println(lists);
    }
    private static void mapRData(final String type, final List<String> head, final DataFrame<Object> result, Observable<ResultScanner> toMap, final boolean addHead)
    {
        toMap.concatMap(new Function<ResultScanner, ObservableSource<Result>>() {
            @Override
            public ObservableSource<Result> apply(ResultScanner results) throws Exception {
                return Observable.fromIterable(results);
            }
        }).map(new Function<Result, List<List<Object>>>() {
            @Override
            public List<List<Object>> apply(Result result) throws Exception {
                Cell[] cells = result.rawCells();
                List<List<Object>> datas = new LinkedList<>();

                String id = CellTool.GetValue(result,COLFAM,ID);
                String rowKey = CellTool.GetRowName(result);
                if (type.equals(SC))
                {
                    for (Cell c:cells)
                    {
                        String key = CellTool.GetColName(c);
                        String value = CellTool.GetValue(c);

                        if (!key.equals(ID) && !key.equals(TYPE))
                        {
                            List<Object> line = emptyList(6);
                            line.set(0,rowKey);
                            line.set(1,id);
                            line.set(5,type);

                            String[] cid_tid_time = key.split("_");
//                            Lg.info(cid_tid_time);
                            if(cid_tid_time.length<2)
                            {
                                continue;
                            }
                            line.set(2,cid_tid_time[0]);
                            if (value.length() == 0)
                            {
                                line.set(3,0);
                            }else
                            {
                                try {

                                    line.set(3,Integer.parseInt(value));
                                }catch (NumberFormatException n)
                                {
                                    line.set(3,0);
                                }
                            }
                            line.set(4,cid_tid_time[1]);
                            datas.add(line);
                        }
                    }
                }else if (type.equals(TC))
                {
                    for (Cell c:cells)
                    {
                        String key = CellTool.GetColName(c);
//                        String value = CellTool.GetValue(c);
                        if (!key.equals(ID) && !key.equals(TYPE))
                        {
                            List<Object> line = emptyList(4);
                            line.set(0,rowKey);
                            line.set(1,id);
                            line.set(2,key);
                            line.set(3,type);
                            datas.add(line);
                        }
                    }
                }

                return datas;
            }
        }).concatMap(new Function<List<List<Object>>, ObservableSource<List<Object>>>() {
            @Override
            public ObservableSource<List<Object>> apply(List<List<Object>> lists) throws Exception {
                return Observable.fromIterable(lists);
            }
        }).subscribe(new Consumer<List<Object>>() {
            @Override
            public void accept(List<Object> strings) throws Exception {
//                Lg.info(strings);
                result.append(strings);
            }
        });

    }

    private static void mapData(final List<String> head, final DataFrame<Object> result, Observable<ResultScanner> toMap, final boolean addHead)
    {
        toMap.concatMap(new Function<ResultScanner, ObservableSource<Result>>() {
            @Override
            public ObservableSource<Result> apply(ResultScanner results) throws Exception {
                return Observable.fromIterable(results);
            }
        }).map(new Function<Result, List<Object>>() {
            @Override
            public List<Object> apply(Result result) throws Exception {
                List<Object> alignData = getAlignData(head, result,addHead);
                return alignData;
            }
        }).subscribe(new Consumer<List<Object>>() {
            @Override
            public void accept(List<Object> strings) throws Exception {
                result.append(strings);
            }
        });
    }
    private static List<Object> emptyList(int len)
    {
        List<Object> result = new LinkedList<>();
        for (int i = 0;i<len;i++)
        {
            result.add("");
        }
        return result;
    }

    public static List<Object> getAlignData(List<String> head,Result result,boolean addHeadInFirst)
    {
        List<Object> data = new ArrayList<>();
        Cell[] cells = result.rawCells();
        for (Cell cell:cells)
        {
            String colName = CellTool.GetColName(cell);
            Object value = CellTool.GetValue(cell);
            if (NUM_COL.contains(colName))
            {
                value = Integer.parseInt(value.toString());
            }
            int index = head.indexOf(colName);
            if (index != -1)
            {
                while (index+1>data.size())
                {
                    data.add("");
                }
                data.set(index,value);
            }
        }
        String rowName = CellTool.GetRowName(result);
        if (addHeadInFirst)
        {
            data.add(0,rowName);

        }else
        {
            data.set(0,rowName);
        }

        return data;
    }

    public static DataFrame<Object> loadAllFromType(String type)
    {
        Lg.info("loadAllFromType(String type)");
        String[] typeLabel = TableProperty.TYPE_MAP.get(type);
        final List<String> cHeader = Arrays.asList(typeLabel);
        final DataFrame<Object> result = new DataFrame<>();
        result.add(typeLabel);
        Observable<ResultScanner> cResult = DataProvider.catchAllFromType(type);
        if (type.equals(SC) || type.equals(TC))
        {
            mapRData(type,cHeader,result,cResult,false);
        }else
        {
            mapData(cHeader,result,cResult,false);
        }

        return result;
    }
}
